package io.wen.bd.s3m3.hbase;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;

public class FriendDeleteProcessor extends BaseRegionObserver {
    private static final Log LOG = LogFactory.getLog(FriendDeleteProcessor.class);

    private static final String FRIENDS_TABLE = "t_friends";
    private static final byte[] FRIENDS_FAMILY = "friends".getBytes();

    /**
     * 删除好友关系
     */
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        LOG.info(">>>> try post delete ...");
        LOG.info(">>>> delete-familyCellMap: " + delete.getFamilyCellMap());

        final Table friendsTable = e.getEnvironment().getTable(TableName.valueOf(FRIENDS_TABLE));
        List<Delete> deleteList = new ArrayList<>();

        final NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        final List<Cell> friendCells = familyCellMap.get(FRIENDS_FAMILY);
        LOG.info(">>>> friendCells: " + friendCells);

        // 删除记录中的列名(quolifier)作为新的要删除行键(rowkey)，可能存在删除多个列的情况
        // 删除记录的行键(rowkey)作为要删除的列名(quolifier)
        for (Cell friendCell : friendCells) {
            final byte[] friendCellQuolifier = CellUtil.cloneQualifier(friendCell);
            final Get get = new Get(friendCellQuolifier);
            get.addColumn(FRIENDS_FAMILY, delete.getRow());
            LOG.info(">>>> check exist: " + friendsTable.exists(get) + " with get: " + get.toJSON());

            // 还存在好友关系则添加到删除列表
            if (friendsTable.exists(get)) {
                LOG.info(">>>> do delete");
                Delete coDelete = new Delete(friendCellQuolifier);
                coDelete.addColumn(FRIENDS_FAMILY, delete.getRow());
                deleteList.add(coDelete);
            }
        }

        LOG.info(">>>> delete list: " + deleteList);
        if (!deleteList.isEmpty()) {
            friendsTable.delete(deleteList);
            LOG.info("delete done.");
        }
        friendsTable.close();
    }
}
